package cn.cs.message;

import java.io.IOException;
import java.net.ConnectException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.BlockedListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ReturnListener;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;

public class ConnectionManage 
{
	private static final Logger logger=LoggerFactory.getLogger(ConnectionManage.class);
	
	private volatile boolean canConnect=true;
	public boolean isCanConnect() {
		return canConnect;
	}

	private ConnectionFactory cf;
	private volatile Connection connection;	
	private int threadPoolNum=0;
	private ExecutorService connConsumerPool;
	
	private ReconnectHandler reconnectHandler;
	/**处理confirm问题*/
	private BlockedListener blockedListener;
	//缓存的信息
	private ShutdownListener shutdownListner=new ShutdownListener() {
		public void shutdownCompleted(ShutdownSignalException cause)
		{
			canConnect=false;
			logger.error("可能rabbitMQ服务器挂掉！",cause);				
			testThread=new Thread(new ConnectLoopTest());
			testThread.start();
		}
	};
	//重新连接循环
	private Thread testThread;
	private class ConnectLoopTest implements Runnable{
		@Override
		public void run() 
		{
			while(!canConnect||!Thread.currentThread().isInterrupted())
			{
				try {
					createConnection();
					canConnect=true;
					logger.info("#########################和rabbitMQ重新连接成功");
					break;
				}catch (ConnectException e2) {		
					logger.error("可能rabbitMQ服务器挂掉！ an error occurred while attempting to connect a socket to a remote address and port. Typically, the connection was refused remotely (e.g., no process is listening on the remote address/port.ConnectException-ConnectLoopTest-run",e2);					
				}catch (ShutdownSignalException e1) {				
					logger.error("可能rabbitMQ服务器挂掉！Depending on HardError when calling com.rabbitmq.client.ShutdownSignalException.getReference() we will either get a reference to the Connection or Channel instance that fired this exception.ShutdownSignalException-ConnectLoopTest-run",e1);					
				}catch (TimeoutException e) {
					logger.error("可能rabbitMQ服务器挂掉！链接超时",e);
				}catch (IOException e) {
					logger.error("可能rabbitMQ服务器挂掉！",e);
				}
				
				try {
					TimeUnit.SECONDS.sleep(30);
				} catch (InterruptedException e) {
					Thread.currentThread().interrupt();
					logger.warn(e.getMessage(),e);
				}			
			}
			
			if(reconnectHandler!=null)
				reconnectHandler.afterReconnect();
		}
		
	}
	
	private void createConnection() throws IOException, TimeoutException
	{
		if(connConsumerPool!=null)
			connConsumerPool.shutdown();
		
		if(threadPoolNum>0)
		{
			this.connConsumerPool=Executors.newFixedThreadPool(threadPoolNum);
			connection=cf.newConnection(connConsumerPool);
		}	
		else 
			connection=cf.newConnection();
		
		connection.addShutdownListener(shutdownListner);
		if(blockedListener!=null)
			connection.addBlockedListener(blockedListener);
	}
	
	public ConnectionManage(String uri,int threadPoolNum,ReconnectHandler handler) throws IOException, TimeoutException, KeyManagementException, NoSuchAlgorithmException, URISyntaxException
	{
		this(uri,threadPoolNum,handler,null);
	}
	
	public ConnectionManage(String uri,ReconnectHandler handler,BlockedListener blockedListener) throws IOException, TimeoutException, KeyManagementException, NoSuchAlgorithmException, URISyntaxException
	{
		this(uri,0,handler,blockedListener);
	}
	
	public ConnectionManage(String uri,int threadPoolNum,ReconnectHandler handler,
							BlockedListener blockedListener) throws IOException, TimeoutException, KeyManagementException, NoSuchAlgorithmException, URISyntaxException
	{
		this.threadPoolNum=threadPoolNum;
		this.reconnectHandler=handler;
		this.blockedListener=blockedListener;
		
		cf=new ConnectionFactory();
		cf.setUri(uri);
		createConnection();
	}
	
	/***创建连接*/
	public Channel createChannel()
	{
		Channel channel=null;
		try {
			channel=connection.createChannel();
		}catch (AlreadyClosedException e2){
			logger.error("createChannel-AlreadyClosedException",e2);
		}catch (ShutdownSignalException e1) {
			logger.error("可能rabbitMQ服务器挂掉！-channel-ShutdownSignalException",e1);					
		}catch (IOException e) {
			logger.error(e.getMessage(),e);
		}
		return channel;
	}
	
	/***创建连接*/
	public Channel createChannel(ReturnListener returnListner)
	{
		Channel channel=null;
		try {
			channel=connection.createChannel();
			channel.addReturnListener(returnListner);
		}catch (AlreadyClosedException e2){
			logger.error("createChannel-AlreadyClosedException",e2);
		}catch (ShutdownSignalException e1) {
			logger.error("可能rabbitMQ服务器挂掉！-channel-ShutdownSignalException",e1);					
		}catch (IOException e) {
			logger.error(e.getMessage(),e);
		}
		return channel;
	}
	
	public void shutdown() throws IOException
	{
		if(testThread!=null)
			testThread.interrupt();
		
		if(connection!=null)
			connection.close();
		
		if(connConsumerPool!=null)
			connConsumerPool.shutdown();
	}
	

	public Connection getConnection() {
		return connection;
	}
	
}
